//package com.zhz.mall.portal.kakfa;
//
//import cn.hutool.core.collection.CollUtil;
//import cn.hutool.core.lang.Assert;
//import com.zhz.mall.portal.PortalApplication;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.aop.framework.autoproxy.AutoProxyUtils;
//import org.springframework.aop.scope.ScopedObject;
//import org.springframework.aop.scope.ScopedProxyUtils;
//import org.springframework.aop.support.AopUtils;
//import org.springframework.beans.factory.BeanInitializationException;
//import org.springframework.beans.factory.SmartInitializingSingleton;
//import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
//import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
//import org.springframework.context.ApplicationContext;
//import org.springframework.context.ApplicationContextAware;
//import org.springframework.context.ConfigurableApplicationContext;
//import org.springframework.core.MethodIntrospector;
//import org.springframework.core.annotation.AnnotatedElementUtils;
//import org.springframework.lang.NonNull;
//import org.springframework.stereotype.Component;
//
//import java.lang.reflect.Method;
//import java.util.Map;
//import java.util.Objects;
//
///**
// * KafkaListenerMethodProcessor
// * <p>
// * 扫描标记有 @KafkaListener 注解的方法, 生成 kafka消息监听器, 分发接收的 kafka 消息
// *
// * @author zhouhengzhe
// * @see KafkaDispatcher
// */
//@Slf4j
//@Component
//public class KafkaListenerMethodProcessor implements SmartInitializingSingleton, ApplicationContextAware, BeanFactoryPostProcessor {
//
//    private static final String PACKAGE_PREFIX = PortalApplication.class.getPackage().getName() + ".";
//
//    private ConfigurableApplicationContext applicationContext;
//    private ConfigurableListableBeanFactory beanFactory;
//    private KafkaDispatcher kafkaDispatcher;
//
//
//    @Override
//    public void setApplicationContext(@NonNull ApplicationContext applicationContext) {
//        Assert.isInstanceOf(ConfigurableApplicationContext.class, applicationContext);
//        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
//    }
//
//
//    @Override
//    public void postProcessBeanFactory(@NonNull ConfigurableListableBeanFactory beanFactory) {
//        this.beanFactory = beanFactory;
//    }
//
//
//    @Override
//    public void afterSingletonsInstantiated() {
//        this.kafkaDispatcher = beanFactory.getBean(KafkaDispatcher.class);
//        ConfigurableListableBeanFactory beanFactory = this.beanFactory;
//
//        String[] beanNames = beanFactory.getBeanNamesForType(Object.class);
//        for (String beanName : beanNames) {
//            if (!ScopedProxyUtils.isScopedTarget(beanName)) {
//                Class<?> type = null;
//                try {
//                    type = AutoProxyUtils.determineTargetClass(beanFactory, beanName);
//                } catch (Throwable ex) {
//                    // An unresolvable bean type, probably from a lazy bean - let's ignore it.
//                    log.info("Could not resolve target class for bean with name '{}'", beanName, ex);
//                }
//
//                if (Objects.nonNull(type)) {
//                    if (ScopedObject.class.isAssignableFrom(type)) {
//                        try {
//                            Class<?> targetClass = AutoProxyUtils.determineTargetClass(
//                                    beanFactory, ScopedProxyUtils.getTargetBeanName(beanName));
//                            if (Objects.nonNull(targetClass)) {
//                                type = targetClass;
//                            }
//                        } catch (Throwable ex) {
//                            // An invalid scoped proxy arrangement - let's ignore it.
//                            if (log.isDebugEnabled()) {
//                                log.debug("Could not resolve target bean for scoped proxy '" + beanName + "'", ex);
//                            }
//                        }
//                    }
//
//                    try {
//                        processBean(beanName, type);
//                    } catch (Throwable ex) {
//                        throw new BeanInitializationException("Failed to process @KafkaListener " +
//                                "annotation on bean with name '" + beanName + "'", ex);
//                    }
//                }
//            }
//        }
//    }
//
//
//    private void processBean(String beanName, Class<?> targetType) {
//        if (!isProjectClass(targetType)) {
//            return;
//        }
//
//        // 查找含有 @KafkaListener 的方法
//        Map<Method, KafkaListener> annotatedMethods;
//        try {
//            annotatedMethods = MethodIntrospector.selectMethods(targetType,
//                    (MethodIntrospector.MetadataLookup<KafkaListener>) method ->
//                            AnnotatedElementUtils.findMergedAnnotation(method, KafkaListener.class));
//        } catch (Throwable ex) {
//            // An unresolvable type in a method signature, probably from a lazy bean - let's ignore it.
//            log.info("Could not resolve methods for bean with name '{}'", beanName, ex);
//            return;
//        }
//
//        if (CollUtil.isEmpty(annotatedMethods)) {
//            return;
//        }
//
//        ConfigurableApplicationContext context = this.applicationContext;
//        annotatedMethods.forEach((method, ann) -> {
//            Method methodToUse = AopUtils.selectInvocableMethod(method, context.getType(beanName));
//            KafkaMessageListener kafkaMessageListener = new DefaultKafkaMessageListener(context, beanName, targetType, methodToUse, ann);
//            kafkaDispatcher.addListener(kafkaMessageListener);
//        });
//
//        log.info("{} @KafkaListener methods processed on bean '{}': {}", annotatedMethods.size(), beanName, annotatedMethods);
//    }
//
//
//    /**
//     * 是否是本工程中的类
//     */
//    private static boolean isProjectClass(Class<?> clazz) {
//        return clazz.getName().startsWith(PACKAGE_PREFIX);
//    }
//
//}
